package com.example;

import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.util.*;

public class SparkDataGenerator {

    private static final String[] CATEGORIES = {"电子产品", "服装", "食品", "家居", "美妆"};
    private static final Random RANDOM = new Random();

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("UserRepurchaseMetrics")
                .setMaster("local[*]");

        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        // 处理文件路径：优先使用传入的参数，否则用默认路径
        String filePath;
        if (args.length > 0) {
            filePath = args[0];
            System.out.println("使用指定的文件路径: " + filePath);
        } else {
            filePath = "data/user.csv";  // 默认路径
            System.out.println("使用默认文件路径: " + filePath);
        }

        // 生成数据的逻辑（不变）
        int numUsers = 1000;
        int purchasesPerUser = 5;
        List<Row> rows = new ArrayList<>();
        // ...（省略数据生成代码）

        // 定义 schema（不变）
        List<StructField> fields = Arrays.asList(
                DataTypes.createStructField("userId", DataTypes.StringType, false),
                DataTypes.createStructField("orderId", DataTypes.StringType, false),
                DataTypes.createStructField("amount", DataTypes.DoubleType, false),
                DataTypes.createStructField("purchaseTime", DataTypes.TimestampType, false),
                DataTypes.createStructField("category", DataTypes.StringType, false)
        );
        StructType schema = DataTypes.createStructType(fields);

        // 创建 DataFrame（不变）
        Dataset<Row> df = spark.createDataFrame(rows, schema);

        // 保存文件：使用处理后的 filePath
        df.write()
                .option("header", "true")
                .csv(filePath);  // 关键修改：用 filePath 替代 args[0]

        // 打印保存路径：同样使用 filePath
        System.out.println("数据生成完成，保存到: " + filePath);  // 关键修改

        spark.stop();
    }

    private static Date getRandomDate() {
        Calendar cal = Calendar.getInstance();
        cal.add(Calendar.DAY_OF_YEAR, -90); // 过去90天内
        cal.add(Calendar.HOUR_OF_DAY, -RANDOM.nextInt(90 * 24));
        return cal.getTime();
    }
}